package com.hejie.videoservice.service.impl;

import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.model.GeneratePresignedUrlRequest;
import com.hejie.common.exception.BusinessException;
import com.hejie.common.exception.ErrorCode;
import com.hejie.common.utils.JsonUtils;
import com.hejie.videoservice.dto.VideoDTO;
import com.hejie.videoservice.dto.VideoPageQueryDTO;
import com.hejie.videoservice.dto.VideoUploadRequest;
import com.hejie.videoservice.entity.Video;
import com.hejie.videoservice.repository.VideoRepository;
import com.hejie.videoservice.service.VideoService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.net.URL;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Service
@Slf4j
public class VideoServiceImpl implements VideoService {

    @Resource
    private VideoRepository videoRepository;

    @Resource
    private AmazonS3 amazonS3;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    // S3存储桶名称
    private static final String BUCKET_NAME = "hejie-video-bucket";
    // Redis缓存前缀
    private static final String VIDEO_CACHE_KEY = "video:info:";
    // 缓存过期时间(小时)
    private static final long CACHE_EXPIRE_HOURS = 24;

    @Override
    @Transactional
    public VideoDTO uploadVideo(VideoUploadRequest uploadRequest) {
        // 1. 生成视频唯一ID
        Long userId = uploadRequest.getUserId();
        String title = uploadRequest.getTitle();
        String description = uploadRequest.getDescription();
        Long categoryId = uploadRequest.getCategoryId();
        Set<String> tags = uploadRequest.getTags();
        Video.Visibility visibility = uploadRequest.getVisibility();

        // 2. 创建视频记录
        Video video = new Video();
        video.setUserId(userId);
        video.setTitle(title);
        video.setDescription(description);
        video.setCategoryId(categoryId);
        video.setCategoryName(uploadRequest.getCategoryName());
        video.setTags(tags != null ? tags : new HashSet<>());
        video.setVisibility(visibility != null ? visibility : Video.Visibility.PUBLIC);
        video.setStatus(Video.VideoStatus.PROCESSING);
        video.setViewCount(0L);
        video.setLikeCount(0L);
        video.setCommentCount(0L);
        video.setFavoriteCount(0L);
        video.setDeleted(false);

        // 3. 保存视频记录
        Video savedVideo = videoRepository.save(video);
        log.info("视频记录保存成功: videoId={}, title={}", savedVideo.getId(), title);

        // 4. 生成S3预签名URL用于视频上传
        String videoKey = String.format("videos/%d/%s.mp4", userId, savedVideo.getId());
        GeneratePresignedUrlRequest urlRequest = new GeneratePresignedUrlRequest(BUCKET_NAME, videoKey);
        urlRequest.setExpiration(new Date(System.currentTimeMillis() + 30 * 60 * 1000)); // 30分钟有效期
        URL uploadUrl = amazonS3.generatePresignedUrl(urlRequest);

        // 5. 发送视频处理消息到Kafka
        Map<String, Object> message = new HashMap<>();
        message.put("videoId", savedVideo.getId());
        message.put("userId", userId);
        message.put("videoKey", videoKey);
        kafkaTemplate.send("video-processing-topic", JsonUtils.toJson(message));

        // 6. 构建并返回DTO
        VideoDTO videoDTO = new VideoDTO();
        BeanUtils.copyProperties(savedVideo, videoDTO);
        videoDTO.setUploadUrl(uploadUrl.toString());

        return videoDTO;
    }

    @Override
    public VideoDTO getVideoById(Long videoId) {
        // 1. 先从缓存获取
        String cacheKey = VIDEO_CACHE_KEY + videoId;
        VideoDTO cachedVideo = (VideoDTO) redisTemplate.opsForValue().get(cacheKey);
        if (cachedVideo != null) {
            return cachedVideo;
        }

        // 2. 缓存未命中，从数据库获取
        Video video = videoRepository.findById(videoId)
                .filter(v -> !v.isDeleted())
                .orElseThrow(() -> new BusinessException(ErrorCode.VIDEO_NOT_FOUND));

        // 3. 检查视频可见性
        checkVideoVisibility(video);

        // 4. 转换为DTO
        VideoDTO videoDTO = convertToDTO(video);

        // 5. 缓存视频信息
        redisTemplate.opsForValue().set(cacheKey, videoDTO, CACHE_EXPIRE_HOURS, TimeUnit.HOURS);

        return videoDTO;
    }

    @Override
    @Transactional
    public VideoDTO updateVideo(Long videoId, Map<String, Object> updateFields) {
        // 1. 获取视频
        Video video = videoRepository.findById(videoId)
                .filter(v -> !v.isDeleted())
                .orElseThrow(() -> new BusinessException(ErrorCode.VIDEO_NOT_FOUND));

        // 2. 检查权限
        if (!video.getUserId().equals(updateFields.get("currentUserId"))) {
            throw new BusinessException(ErrorCode.NO_PERMISSION);
        }

        // 3. 更新字段
        if (updateFields.containsKey("title")) {
            video.setTitle((String) updateFields.get("title"));
        }
        if (updateFields.containsKey("description")) {
            video.setDescription((String) updateFields.get("description"));
        }
        if (updateFields.containsKey("categoryId")) {
            video.setCategoryId((Long) updateFields.get("categoryId"));
            video.setCategoryName((String) updateFields.get("categoryName"));
        }
        if (updateFields.containsKey("tags")) {
            video.setTags((Set<String>) updateFields.get("tags"));
        }
        if (updateFields.containsKey("visibility")) {
            video.setVisibility((Video.Visibility) updateFields.get("visibility"));
        }

        // 4. 保存更新
        Video updatedVideo = videoRepository.save(video);
        log.info("视频更新成功: videoId={}", videoId);

        // 5. 更新缓存
        String cacheKey = VIDEO_CACHE_KEY + videoId;
        VideoDTO videoDTO = convertToDTO(updatedVideo);
        redisTemplate.opsForValue().set(cacheKey, videoDTO, CACHE_EXPIRE_HOURS, TimeUnit.HOURS);

        return videoDTO;
    }

    @Override
    @Transactional
    public boolean deleteVideo(Long videoId) {
        // 1. 获取视频
        Video video = videoRepository.findById(videoId)
                .filter(v -> !v.isDeleted())
                .orElseThrow(() -> new BusinessException(ErrorCode.VIDEO_NOT_FOUND));

        // 2. 逻辑删除视频
        int rows = videoRepository.logicallyDeleteVideo(videoId);
        if (rows > 0) {
            log.info("视频删除成功: videoId={}", videoId);

            // 3. 删除缓存
            redisTemplate.delete(VIDEO_CACHE_KEY + videoId);

            // 4. 发送视频删除消息到Kafka
            Map<String, Object> message = new HashMap<>();
            message.put("videoId", videoId);
            message.put("videoUrl", video.getVideoUrl());
            message.put("thumbnailUrl", video.getThumbnailUrl());
            kafkaTemplate.send("video-delete-topic", JsonUtils.toJson(message));

            return true;
        }
        return false;
    }

    @Override
    public Page<VideoDTO> getUserVideos(Long userId, Pageable pageable) {
        Page<Video> videoPage = videoRepository.findByUserIdAndDeletedFalse(userId, pageable);
        return videoPage.map(this::convertToDTO);
    }

    @Override
    public Page<VideoDTO> getCategoryVideos(Long categoryId, Pageable pageable) {
        Page<Video> videoPage = videoRepository.findByCategoryIdAndStatusAndDeletedFalse(
                categoryId, Video.VideoStatus.PUBLISHED, pageable);
        return videoPage.map(this::convertToDTO);
    }

    @Override
    public Page<VideoDTO> searchVideos(String keyword, Pageable pageable) {
        Page<Video> videoPage = videoRepository.searchVideos(
                keyword, Video.VideoStatus.PUBLISHED, Video.Visibility.PUBLIC, pageable);
        return videoPage.map(this::convertToDTO);
    }

    @Override
    public Page<VideoDTO> getHotVideos(Pageable pageable) {
        Page<Video> videoPage = videoRepository.findHotVideos(
                Video.VideoStatus.PUBLISHED, Video.Visibility.PUBLIC, pageable);
        return videoPage.map(this::convertToDTO);
    }

    @Override
    public Page<VideoDTO> getLatestVideos(Pageable pageable) {
        Page<Video> videoPage = videoRepository.findLatestVideos(
                Video.VideoStatus.PUBLISHED, Video.Visibility.PUBLIC, pageable);
        return videoPage.map(this::convertToDTO);
    }

    @Override
    public List<VideoDTO> batchGetVideos(List<Long> ids) {
        if (ids == null || ids.isEmpty()) {
            return Collections.emptyList();
        }

        List<Video> videos = videoRepository.findByIdInAndDeletedFalse(ids);
        return videos.stream()
                .map(this::convertToDTO)
                .collect(Collectors.toList());
    }

    @Override
    @Transactional
    public VideoDTO updateVideoStatus(Long videoId, Video.VideoStatus status) {
        // 1. 获取视频
        Video video = videoRepository.findById(videoId)
                .filter(v -> !v.isDeleted())
                .orElseThrow(() -> new BusinessException(ErrorCode.VIDEO_NOT_FOUND));

        // 2. 更新状态
        video.setStatus(status);
        Video updatedVideo = videoRepository.save(video);

        // 3. 更新缓存
        String cacheKey = VIDEO_CACHE_KEY + videoId;
        VideoDTO videoDTO = convertToDTO(updatedVideo);
        redisTemplate.opsForValue().set(cacheKey, videoDTO, CACHE_EXPIRE_HOURS, TimeUnit.HOURS);

        // 4. 如果是发布状态，发送视频发布事件
        if (status == Video.VideoStatus.PUBLISHED) {
            Map<String, Object> message = new HashMap<>();
            message.put("videoId", videoId);
            message.put("userId", video.getUserId());
            message.put("title", video.getTitle());
            kafkaTemplate.send("video-publish-topic", JsonUtils.toJson(message));
        }

        return videoDTO;
    }

    @Override
    @Transactional
    public boolean incrementViewCount(Long videoId) {
        // 1. 更新观看次数
        int rows = videoRepository.incrementViewCount(videoId);
        if (rows <= 0) {
            return false;
        }

        // 2. 更新缓存
        String cacheKey = VIDEO_CACHE_KEY + videoId;
        VideoDTO cachedVideo = (VideoDTO) redisTemplate.opsForValue().get(cacheKey);
        if (cachedVideo != null) {
            cachedVideo.setViewCount(cachedVideo.getViewCount() + 1);
            redisTemplate.opsForValue().set(cacheKey, cachedVideo, CACHE_EXPIRE_HOURS, TimeUnit.HOURS);
        }

        return true;
    }

    @Override
    public boolean updateLikeCount(Long videoId, int increment) {
        // 1. 更新点赞数
        int rows = videoRepository.updateLikeCount(videoId, increment);
        if (rows <= 0) {
            return false;
        }

        // 2. 更新缓存
        String cacheKey = VIDEO_CACHE_KEY + videoId;
        VideoDTO cachedVideo = (VideoDTO) redisTemplate.opsForValue().get(cacheKey);
        if (cachedVideo != null) {
            cachedVideo.setLikeCount(cachedVideo.getLikeCount() + increment);
            redisTemplate.opsForValue().set(cacheKey, cachedVideo, CACHE_EXPIRE_HOURS, TimeUnit.HOURS);
        }

        return true;
    }

    @Override
    public boolean updateCommentCount(Long videoId, int increment) {
        // 1. 更新评论数
        int rows = videoRepository.updateCommentCount(videoId, increment);
        if (rows <= 0) {
            return false;
        }

        // 2. 更新缓存
        String cacheKey = VIDEO_CACHE_KEY + videoId;
        VideoDTO cachedVideo = (VideoDTO) redisTemplate.opsForValue().get(cacheKey);
        if (cachedVideo != null) {
            cachedVideo.setCommentCount(cachedVideo.getCommentCount() + increment);
            redisTemplate.opsForValue().set(cacheKey, cachedVideo, CACHE_EXPIRE_HOURS, TimeUnit.HOURS);
        }

        return true;
    }

    @Override
    public Page<VideoDTO> queryVideos(VideoPageQueryDTO queryDTO, Pageable pageable) {
        Page<Video> videoPage;

        if (queryDTO.getUserId() != null) {
            // 查询用户视频
            videoPage = videoRepository.findByUserIdAndDeletedFalse(queryDTO.getUserId(), pageable);
        } else if (queryDTO.getCategoryId() != null) {
            // 查询分类视频
            videoPage = videoRepository.findByCategoryIdAndStatusAndDeletedFalse(
                    queryDTO.getCategoryId(), Video.VideoStatus.PUBLISHED, pageable);
        } else if (queryDTO.getKeyword() != null && !queryDTO.getKeyword().isEmpty()) {
            // 搜索视频
            videoPage = videoRepository.searchVideos(
                    queryDTO.getKeyword(), Video.VideoStatus.PUBLISHED, Video.Visibility.PUBLIC, pageable);
        } else if ("hot".equals(queryDTO.getSortType())) {
            // 热门视频
            videoPage = videoRepository.findHotVideos(
                    Video.VideoStatus.PUBLISHED, Video.Visibility.PUBLIC, pageable);
        } else {
            // 默认最新视频
            videoPage = videoRepository.findLatestVideos(
                    Video.VideoStatus.PUBLISHED, Video.Visibility.PUBLIC, pageable);
        }

        return videoPage.map(this::convertToDTO);
    }

    /**
     * 转换为DTO
     */
    private VideoDTO convertToDTO(Video video) {
        VideoDTO dto = new VideoDTO();
        BeanUtils.copyProperties(video, dto);
        return dto;
    }

    /**
     * 检查视频可见性
     */
    private void checkVideoVisibility(Video video) {
        // 这里应该根据当前登录用户ID和视频可见性做权限检查
        // 简化实现，只检查视频状态
        if (video.getStatus() != Video.VideoStatus.PUBLISHED) {
            throw new BusinessException(ErrorCode.VIDEO_NOT_AVAILABLE);
        }
    }
}